package com.captjack.bigdata.storm.topology;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.topology.TopologyBuilder;

/**
 * 使用“LocalCluster”对象创建本地集群，然后使用“LocalCluster”类的“submitTopology”方法提交拓扑。
 * “submitTopology”的参数之一是“Config”类的实例。“Config”类用于在提交拓扑之前设置配置选项。
 * 此配置选项将在运行时与集群配置合并，并使用prepare方法发送到所有任务（spout和bolt）。
 * 一旦拓扑提交到集群，我们将等待10秒钟，集群计算提交的拓扑，然后使用“LocalCluster”的“shutdown”方法关闭集群。
 *
 * @author Jack Sparrow
 * @version 1.0.0
 * @date 2018/4/13 01:04
 * package com.cc.bigdata.storm.topology
 */
public class StreamDataTopology {

    public static void main(String[] args) {
        try {
            Config config = new Config();
            config.setDebug(true);
            //
            TopologyBuilder topologyBuilder = new TopologyBuilder();
            // 设置生产者
            topologyBuilder.setSpout("call-log-reader-spout", new StreamDataSpout());
            // 设置消费者
            topologyBuilder.setBolt("call-log-creator-bolt", new StreamDataBolt())
                    .shuffleGrouping("call-log-reader-spout");
            LocalCluster cluster = new LocalCluster();
            // 将拓扑提交到storm集群
//            StormSubmitter.submitTopology("LogAnalyserStorm", config, topologyBuilder.createTopology());
            cluster.submitTopology("LogAnalyserStorm", config, topologyBuilder.createTopology());
            Thread.sleep(10000);
            // 终止此拓扑
            cluster.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
